EC2インスタンス起動時にInsufficientInstanceCapacityが発生したら別のインスタンスタイプで起動させてみる

こんにちは、muroです。AWS運用かんたん自動化ツール「opswitch」の開発・運用を担当しています。opswitchはboto3をがんがん使ってAWSを操作します。今回はboto3のモックライブラリ「moto」を利用して、EC2インスタンスの起動をテストしてみます。ただし、EC2インスタンスを起動するだけではなく、よくあるトラブルに対応した実装を試します。

前回記事はこちら。

boto3からEC2へのアクセスをmotoでモックしてみる

シナリオ

EC2インスタンス起動時にInsufficientInstanceCapacityエラーが発生し、インスタンスを起動できないことがあります。

EC2 インスタンスの開始または起動時に発生する、InsufficientInstanceCapacity エラーのトラブルシューティング - AWS

数回リトライしても起動できなかった場合、別のインスタンスタイプに変更して起動させるようにしてみます。

環境

今回のテストに使用した環境は以下の通りです。

環境	バージョン
Python	3.9.13
boto3	1.23.3
moto	3.1.9
pytest	7.1.2
pytest-mock	3.7.0
retrying	1.3.3

またテストプロジェクトの構成は以下の通りです。

.
├── main.py
└── test_main.py

やってみる

テスト対象コード

import boto3
from retrying import retry
from botocore.exceptions import ClientError
import logging

logger = logging.getLogger()


def start_instances(instance_ids):
    """
    EC2インスタンスを起動します。
    """
    ec2 = boto3.client('ec2')
    ec2.start_instances(
        InstanceIds=instance_ids
    )


def change_instance_type(instance_id, alt_instance_type):
    """
    EC2インスタンスのインスタンスタイプを変更します。
    """
    ec2 = boto3.client('ec2')
    ec2.modify_instance_attribute(
        InstanceId=instance_id,
        Attribute='instanceType',
        Value=alt_instance_type
    )


@retry(wait_exponential_multiplier=1000, stop_max_attempt_number=5)
def start_instance(instance_id):
    """
    EC2インスタンスの起動をリトライするラッパー関数です。
    """
    logger.info('try start_instances...')
    start_instances([instance_id])


def force_start_instance(instance_id, alt_instance_type):
    """
    EC2インスタンスを起動します。
    InsufficientInstanceCapacityエラー時はインスタンスタイプを変更して
    再度EC2インスタンスの起動を試みます。
    """
    try:
        start_instance(instance_id)
    except ClientError as ce:
        logger.error(ce)
        error_code = ce.response['Error']['Code']
        if error_code == 'InsufficientInstanceCapacity':
            change_instance_type(instance_id, alt_instance_type)
            start_instance(instance_id)

retryingについて

retrying · PyPI

関数に@retryデコレータを指定するだけでその関数をリトライしてくれる便利なライブラリです。今回はエクスポネンシャル・バックオフで時間をおいて最大5回リトライするように指定しました。retryingの詳しい使い方は以下の記事を参考にしてください。

PythonでExponential Backoffをしたかったのでretryingモジュールを調べてみた

エラーハンドリングについて

botocore.exceptions.ClientError をキャッチして、エラーコードがInsufficientInstanceCapacityエラーかどうかを判定しています。インスタンスタイプを変更して、さらにstart_instance()を呼び出しますが、それでも起動できなかった場合は例外が発生して終了します。boto3のエラーハンドリングについては以下のリファレンスも参考にしてください。

Error handling — Boto3 Docs 1.24.61 documentation

テストコード

今回もmotoを利用してEC2インスタンスをモックしています。さらに、pytest-mock を利用して、start_instances関数をモックしています。インスタンスタイプが代替指定のインスタンスタイプでない場合InsufficientInstanceCapacityエラーを発生させます。インスタンスタイプが代替指定のインスタンスタイプの場合にEC2インスタンスを起動して、結果をassertします。

import boto3
from moto import mock_ec2
import main
from botocore.exceptions import ClientError
import logging

logger = logging.getLogger()


@mock_ec2
def test_force_start_instances(mocker):
    expected_number_of_instances = 1
    initial_instance_type = 't2.medium'
    input_alt_instance_type = 't3.medium'
    input_instance_id = run_instances(
        expected_number_of_instances,
        initial_instance_type
    )[0]

    # 例外を発生させるようにMockを設定します。
    def mock_start_instances(instance_ids):
        instance = describe_instance(instance_ids[0])
        instance_type = instance['InstanceType']
        if instance_type == input_alt_instance_type:
            ec2 = boto3.client('ec2')
            ec2.start_instances(
                InstanceIds=instance_ids
            )
        else:
            raise ClientError(
                {'Error': {'Code': 'InsufficientInstanceCapacity'}},
                'StartInstances'
            )
    mocker.patch('main.start_instances', side_effect=mock_start_instances)

    # テストを実行します。
    main.force_start_instance(input_instance_id, input_alt_instance_type)

    # インスタンスが起動されたかどうかテストします。
    instance = describe_instance(input_instance_id)
    assert(instance['InstanceType'] == input_alt_instance_type)
    assert(instance['State']['Name'] == 'running')


def run_instances(expected_number_of_instances, instance_type):
    # 期待するインスタンス数の分だけインスタンスを起動します。
    ec2 = boto3.client('ec2')
    images = ec2.describe_images()['Images']
    response = ec2.run_instances(
        ImageId=images[0]['ImageId'],
        MinCount=expected_number_of_instances,
        MaxCount=expected_number_of_instances,
        InstanceType=instance_type
    )
    # インスタンスIDのリストを取得します。
    instance_ids = [x['InstanceId'] for x in response['Instances']]
    # インスタンスを停止します。
    ec2.stop_instances(
        InstanceIds=instance_ids
    )
    # インスタンスIDのリストを返します。
    return instance_ids


def describe_instance(instance_id):
    ec2 = boto3.client('ec2')
    response = ec2.describe_instances(InstanceIds=[instance_id])
    return response['Reservations'][0]['Instances'][0]

テスト実行

それではテストを実行してみます。今回のテスト対象コードはloggerを仕込んでいるので、その出力を表示させながら実行します。

python -m pytest --log-cli-level=INFO -s  \
--log-cli-format="%(asctime)s.%(msecs)03d %(levelname)s %(message)s"
========================================== test session starts ==========================================
platform linux -- Python 3.9.13, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/ユーザー/プロジェクト
plugins: cov-3.0.0, mock-3.7.0
collected 1 item

test_main.py::test_force_start_instances 
--------------------------------------------- live log call ---------------------------------------------
17:57:01.855 INFO Found credentials in environment variables.
17:57:02.041 INFO try start_instances...
17:57:04.072 INFO try start_instances...
17:57:08.139 INFO try start_instances...
17:57:16.193 INFO try start_instances...
17:57:32.242 INFO try start_instances...
17:57:32.299 ERROR An error occurred (InsufficientInstanceCapacity) when calling the StartInstances operation: Unknown
17:57:32.308 INFO try start_instances...
PASSED

========================================== 1 passed in 31.20s ===========================================

2秒・4秒・8秒・16秒とリトライ間隔をあけてリトライされ、エラーの後テストが成功しました。

まとめ

retryingライブラリを利用することでエクスポネンシャル・バックオフで待機時間を延ばしながら関数をリトライさせることができました。また、motoでEC2インスタンスをモックしつつpytest-mockを利用して複雑なエラー条件をテストすることができました。なかなか発生させにくいエラーとそのハンドリングであっても、ライブラリを利用することで、ある程度再現させられるのはとても便利だと思いました。以上、ご参考になりましたら幸いです。